package com.bbx.flink.demo.evictors;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.evictors.TimeEvictor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

@Slf4j
public class TimeEvictorDemo {


    /**
     * TimeEvictor    时间驱逐  ------运行结果未达到预期，不太明白
     * @param args
     * @throws Exception
     */
    public static void main(String [] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        TimeEvictor<TimeWindow> timeEvictor = TimeEvictor.of(Time.milliseconds(1000));
        long windowSize = timeEvictor.getWindowSize();
        log.info("windowSize[{}]",windowSize);
        env.socketTextStream("192.168.10.131", 10003)
                .filter((FilterFunction<String>) val -> {
                    log.info("~~~~~~~~~~~~val[{}]", val, System.currentTimeMillis());
                    return true;

                })
                .map((MapFunction<String, Integer>) vaiue -> Integer.parseInt(vaiue))
                .keyBy(i -> i % 2 == 0)
                .timeWindow(Time.minutes(1L))
                .evictor(timeEvictor)
                .sum(0)
                .print();
        env.execute();
    }
}
